package cn.mq.service.impl;

import cn.mq.service.MqService;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;

/**
 * 单次消费Mq实现
 *
 * @author zhanghui145
 * @date 2022/1/14 9:48
 * @since 1.0.0
 */
@Service
public class OnceMqServiceImpl implements MqService, MessageListener {

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private RedisMessageListenerContainer listenerContainer;

    @Resource
    private JdkSerializationRedisSerializer jdkSerializationRedisSerializer;

    // 队列
    Map<String, List<BiConsumer<String, String>>> queueMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        // new PatternTopic("MQ*"), 表示消息匹配规则，MQ*表示只匹配topic为MQ开头的消息
        listenerContainer.addMessageListener(this, new PatternTopic("mq*"));
    }

    @Override
    public void send(String topic, String message) {
        redisTemplate.convertAndSend(topic, message);
    }

    @Override
    public void subscribe(String topic, BiConsumer<String, String> consumer) {
        queueMap.computeIfAbsent(topic, key -> new ArrayList<>()).add(consumer);
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 需要用redis的反序列化，不然会有乱码问题
        String msg = (String) jdkSerializationRedisSerializer.deserialize(message.getBody());
        String topic = new String(message.getChannel());
        queueMap.getOrDefault(topic, new ArrayList<>()).forEach(subscriber -> subscriber.accept(topic, msg));
    }
}
